use super::{
    err, err_value, json, ok, throw_err, Base, CSend, Client, Content, ECode, Fbf, KeyNode, Lock,
    MapSv, RNull, Resp, SStr, SvCache, VBase, VMap, Value, VecS, VecSv, VecV,
};
use std::collections::HashSet;

pub struct Collect<'a> {
    client: &'a Client,
    cache: SvCache,
    line: &'a str,
    reader: Fbf,
}

impl Base for Collect<'_> {}
impl Content for Collect<'_> {}

impl<'a> Collect<'a> {
    pub async fn do_cmd(
        &self,
        tp: &str,
        cmd: &str,
        data: &Value,
        lock: Lock,
        count: Option<VecS>,
    ) -> Resp {
        match cmd {
            "get" => {
                let sn = data.ck_str("k")?;
                let body = self.item_sn(tp, sn).await?;
                ok!(body)
            }
            "list" => {
                let dsn = self.key_all(tp)?;
                let body = self.list_array(&dsn, tp, count).await?;
                ok!(body)
            }
            "add" => {
                let dsn = self.key_all(tp)?;

                let _ = lock.lock().await;
                self.add_item_sub(tp, &dsn, data).await?;

                ok!()
            }
            "madd" => {
                let dsn = self.key_all(tp)?;
                let val = data.ck_array("v")?;

                let _ = lock.lock().await;

                let mut put = Vec::new();
                let mut sn_vec = Vec::new();

                for d in val.iter() {
                    let d = d.ckobj()?;

                    let sn = d.ck_str("sn")?;
                    let k_sn = self.key_sn(tp, sn)?;

                    self.merge(&k_sn, d, &mut put)?;

                    sn_vec.push(json!(sn));
                }

                self.merge_mass_vec(&dsn, &sn_vec, &mut put)?;

                self.commit(Some(&put), None).await?;

                ok!()
            }
            "mditem" => {
                let _ = lock.lock().await;

                let val = data.ck_val("v")?;
                let sn = val.ck_str("sn")?;

                self.md_item(tp, sn, val).await?;

                ok!()
            }
            "md" => {
                let _ = lock.lock().await;
                self.md_item_set(tp, data).await?;

                ok!()
            }
            "rm" => {
                let _ = lock.lock().await;
                self.rm_item(tp, data).await?;

                ok!()
            }
            _ => throw_err!(ECode::NoServe, format!("no collect {tp} serve")),
        }
    }

    pub async fn do_sub_cmd(
        &self,
        tp: &str,
        cmd: &str,
        para: &Value,
        sub: &str,
        lock: Lock,
    ) -> Resp {
        let data = para.ckobj()?;
        let sn = data.ck_str("k")?;

        if cmd.starts_with("add") {
            let val = data.ck_array("v")?;
            let dsn = self.key_sub_sn(tp, sub, sn)?;

            let _ = lock.lock().await;

            let mut put = Vec::new();
            self.merge_mass_vec(&dsn, val, &mut put)?;

            self.commit(Some(&put), None).await?;

            ok!()
        } else if cmd.starts_with("rm") {
            let item = data.ck_str("sn")?;
            let dsn = self.key_sub_sn(tp, sub, sn)?;

            let _ = lock.lock().await;

            let mut put = Vec::new();
            let mut del = Vec::new();

            self.del_mass_item(&dsn, item, &mut put, &mut del);
            self.commit(Some(&put), Some(&del)).await?;

            ok!()
        } else if cmd.starts_with("get") {
            let body = self.item_sub_sn(tp, sub, sn).await?;
            ok!(body)
        } else {
            throw_err!(ECode::NoServe, "no collect sub {tp} serve");
        }
    }

    async fn rm_item(&self, tp: &str, data: &Value) -> RNull {
        let sn = data.ck_str("k")?.to_string();
        let dsn = self.key_all(tp)?;

        let mut put = Vec::new();
        let mut del = vec![self.key_sn(tp, &sn)?];

        let sub_list = vec!["user", "psn", "rework"];

        for d in sub_list.iter() {
            let k = self.key_sub_sn(tp, d, &sn)?;
            del.push(k);
        }

        self.dir_del(&self.cache, self.client, &dsn, &sn, &mut put, &mut del);

        self.commit(Some(&put), Some(&del)).await
    }

    pub async fn del_station_psn(
        &self,
        set: &HashSet<String>,
        put: &mut VecSv,
        del: &mut VecS,
    ) -> RNull {
        if let Ok(station) = self.content_all("station").await {
            for d in station.iter() {
                let k_psn = self.key_sub_sn("station", "psn", d)?;
                let k_rework = self.key_sub_sn("station", "rework", d)?;

                self.del_mass_set(&k_psn, &set, false, put, del);
                self.del_mass_set(&k_rework, &set, false, put, del);
            }
        }

        Ok(())
    }

    pub fn del_run_item(&self, dsn: &str, sn: &str, put: &mut VecSv, del: &mut VecS) -> bool {
        self.run_del(&self.cache, self.client, dsn, sn, put, del)
    }

    pub async fn del_run_set(&self, set: &SStr, tx: &CSend) -> RNull {
        let k_run = self.key_run("task")?;

        let mut put = Vec::new();
        let mut del = Vec::new();

        self.del_mass_set(&k_run, set, true, &mut put, &mut del);

        self.commit_que(Some(&put), Some(&del), tx).await
    }

    pub fn del_mass_item(&self, dsn: &str, sn: &str, put: &mut VecSv, del: &mut VecS) -> bool {
        let item = sn.to_string();
        self.dir_del(&self.cache, self.client, dsn, &item, put, del)
    }

    pub fn del_mass_set(
        &self,
        dsn: &str,
        set: &SStr,
        run: bool,
        put: &mut VecSv,
        del: &mut VecS,
    ) -> bool {
        if run {
            self.run_del_set(&self.cache, self.client, dsn, set, put, del)
        } else {
            self.dir_del_set(&self.cache, self.client, dsn, set, put, del)
        }
    }

    pub async fn del_mass(
        &self,
        sub_tp: Option<&str>,
        dsn: &str,
        del: &mut VecS,
    ) -> Result<SStr, Value> {
        let mut ret = SStr::new();

        if let Ok(body) = self.content(&dsn).await {
            match sub_tp {
                Some(tp) => {
                    for d in body.iter() {
                        let k_item = self.key_sn(tp, d)?;

                        del.push(k_item);
                        ret.insert(d.to_string());
                    }
                }
                None => {
                    for d in body.iter() {
                        let k_item = format!("{dsn}{d}");

                        del.push(k_item);
                        ret.insert(d.to_string());
                    }
                }
            }
        }

        del.push(dsn.to_string());

        Ok(ret)
    }

    pub async fn add_set_to_mass(&self, dsn: &str, val: &MapSv) -> RNull {
        let mut put = Vec::new();

        self.dir_add_set(&self.cache, self.client, dsn, val, &mut put)?;

        self.commit(Some(&put), None).await
    }

    pub fn rm_field(&self, sn: &str, val: &VecV, put: &mut VecSv) -> RNull {
        self.rm_key(&self.cache, self.client, sn, val, put)
    }

    pub fn merge(&self, sn: &str, val: &MapSv, put: &mut VecSv) -> RNull {
        self.merge_item(&self.cache, self.client, sn, val, put)
    }

    pub fn merge_mass(&self, dsn: &str, sn: &str, put: &mut VecSv) {
        self.dir_add(&self.cache, self.client, dsn, sn, put)
    }

    pub fn merge_mass_vec(&self, sn: &str, val: &VecV, put: &mut VecSv) -> RNull {
        self.dir_add_vec(&self.cache, self.client, sn, val, put)
    }

    pub async fn commit(&self, put: Option<&VecSv>, del: Option<&VecS>) -> RNull {
        let ret = self.client.batch(put, del)?;

        if let Some(body) = put {
            for d in body.iter() {
                self.cache.insert(d.0.clone(), d.1.clone()).await;
            }
        }

        if let Some(body) = del {
            for d in body.iter() {
                self.cache.invalidate(d).await;
            }
        }

        Ok(ret)
    }

    pub async fn commit_que(&self, put: Option<&VecSv>, del: Option<&VecS>, tx: &CSend) -> RNull {
        let ret = self.client.put_que(put, del, tx).await?;

        if let Some(body) = put {
            for d in body.iter() {
                self.cache.insert(d.0.clone(), d.1.clone()).await;
            }
        }

        if let Some(body) = del {
            for d in body.iter() {
                self.cache.invalidate(d).await;
            }
        }

        Ok(ret)
    }

    pub async fn add_item_sub(&self, tp: &str, dsn: &str, para: &Value) -> RNull {
        let val = para.ck_obj("v")?;

        let sn = val.ck_str_empty_trim("sn")?;
        let k_item = self.key_sn(tp, sn)?;

        let mut put = Vec::new();

        self.merge(&k_item, val, &mut put)?;
        self.merge_mass(dsn, sn, &mut put);

        self.commit(Some(&put), None).await
    }

    pub async fn md_item_set(&self, tp: &str, data: &Value) -> RNull {
        let val = data.ck_array("v")?;

        let mut ignore = vec!["sn".to_string()];

        if let Some(ex) = data.k_array("ex") {
            for d in ex.iter() {
                let d = d.ckstr()?;
                ignore.push(d.to_string());
            }
        }

        let mut put = Vec::new();

        for d in val.iter() {
            let mut d = d.ckobj()?.clone();

            let sn = d.ck_str("sn")?;
            let dsn = self.key_sn(tp, sn)?;

            for t in ignore.iter() {
                d.remove(t);
            }

            self.merge(&dsn, &d, &mut put)?;
        }

        self.commit(Some(&put), None).await
    }

    pub async fn md_item(&self, tp: &str, sn: &str, data: &Value) -> RNull {
        let k_sn = self.key_sn(tp, sn)?;
        let val = data.ckobj()?;

        if let Some(new_sn) = val.k_str("sn") {
            if sn != new_sn {
                throw_err!(ECode::Para, "sn not the same");
            }
        }

        let mut put = Vec::new();
        self.merge(&k_sn, &val, &mut put)?;

        self.commit(Some(&put), None).await
    }

    pub async fn list_array_ext(&self, tp: &str, sub: &str, dsn: &str, sn: &str) -> Resp {
        let body = self.content(&dsn).await?;

        let mut ret = Vec::new();

        for d in body.iter() {
            let k_item = self.key_sub_sn_ext(tp, sub, sn, d)?;

            if let Ok(v) = self.item(&k_item).await {
                ret.push(v);
            }
        }

        if ret.is_empty() {
            throw_err!(ECode::EmptyData, "no data");
        }

        Ok(json!(ret))
    }

    pub async fn list_obj(&self, tp: &str, sub: &str, dsn: &str, sn: &str) -> Resp {
        let body = self.content(&dsn).await?;

        let mut ret = MapSv::new();

        for d in body.iter() {
            let k_item = self.key_sub_sn_ext(tp, sub, sn, d)?;

            if let Ok(v) = self.item(&k_item).await {
                ret.insert(d.to_string(), v.clone());
            }
        }

        if ret.is_empty() {
            throw_err!(ECode::EmptyData, "no data");
        }

        Ok(json!(ret))
    }

    pub async fn list_array(&self, dsn: &str, tp: &str, count: Option<VecS>) -> Resp {
        let body = self.content(&dsn).await?;

        let mut ret = Vec::new();

        for d in body.iter() {
            if let Some(ref ct) = count {
                if let Ok(mut item) = self.item_sn(tp, d).await {
                    let item = item.ckobj_mut()?;

                    for c in ct.iter() {
                        if let Ok(sub) = self.item_sub_sn(tp, c, d).await {
                            if let Some(sub) = sub.as_array() {
                                item.insert(format!("{c}_num"), json!(sub.len()));
                            }
                        }
                    }

                    ret.push(json!(item));
                }
            } else if let Ok(item) = self.item_sn(tp, d).await {
                ret.push(item);
            }
        }

        if ret.is_empty() {
            throw_err!(ECode::EmptyData, "no data");
        }

        Ok(json!(ret))
    }

    pub async fn list_array_sub(&self, dsn: &str, tp: &str, sub: &str) -> Resp {
        let body = self.content(&dsn).await?;

        let mut ret = Vec::new();

        for d in body.iter() {
            if let Ok(v) = self.item_sub_sn(tp, sub, d).await {
                ret.push(v);
            }
        }

        if ret.is_empty() {
            throw_err!(ECode::EmptyData, "no data");
        }

        Ok(json!(ret))
    }

    pub async fn item_sn(&self, tp: &str, sn: &str) -> Resp {
        let dsn = self.key_sn(tp, sn)?;
        self.item(&dsn).await
    }

    pub async fn item_sub_sn(&self, tp: &str, sub: &str, sn: &str) -> Resp {
        let dsn = self.key_sub_sn(tp, sub, sn)?;
        self.item(&dsn).await
    }

    pub async fn item(&self, dsn: &str) -> Resp {
        let mut is_buffer = false;
        let body = self.item_content(&self.cache, self.client, dsn, &mut is_buffer)?;

        if !is_buffer {
            self.cache.insert(dsn.to_string(), body.clone()).await;
        }

        Ok(body)
    }

    pub async fn content_sub(&self, tp: &str, sub: &str, sn: &str) -> Result<VecS, Value> {
        let dsn = self.key_sub_sn(tp, sub, sn)?;
        self.content(&dsn).await
    }

    pub async fn content_all(&self, tp: &str) -> Result<VecS, Value> {
        let dsn = self.key_all(tp)?;
        self.content(&dsn).await
    }

    pub async fn content_run(&self, tp: &str) -> Result<VecS, Value> {
        let dsn = self.key_run(tp)?;
        self.content(&dsn).await
    }

    pub async fn content(&self, dsn: &str) -> Result<VecS, Value> {
        let mut is_buffer = false;
        let body = self.dir_content(&self.cache, self.client, dsn, &mut is_buffer)?;

        if !is_buffer {
            self.cache.insert(dsn.to_string(), json!(body)).await;
        }

        Ok(body)
    }

    pub fn new(client: &'a Client, cache: SvCache, line: &'a str) -> Self {
        Collect {
            client,
            cache,
            line,
            reader: Fbf,
        }
    }

    pub fn get_file(&self, sub: &str, dsn: &str) -> Resp {
        self.reader.get_file(self.line, sub, dsn, &self.cache)
    }

    pub async fn update_file(&self, sub: &str, dsn: &str, data: &Value) -> RNull {
        self.reader
            .update_file(self.line, sub, dsn, data, &self.cache)
            .await
    }

    pub fn save_file(&self, sub: &str, k: &str, v: &Value) -> Result<(), Value> {
        self.reader.save_file(self.line, sub, k, v)
    }
}

impl KeyNode for Collect<'_> {
    fn key_run(&self, model: &str) -> Result<String, Value> {
        self.key_value(self.line, model, Some("run"), "", None)
    }

    fn key_pause(&self, model: &str) -> Result<String, Value> {
        self.key_value(self.line, model, Some("pause"), "", None)
    }

    fn key_all(&self, model: &str) -> Result<String, Value> {
        self.key_value(self.line, model, Some("all"), "", None)
    }

    fn key_sn(&self, model: &str, sn: &str) -> Result<String, Value> {
        self.key_value(self.line, model, None, sn, None)
    }

    fn key_sn_ext(&self, model: &str, sn: &str, ext: &str) -> Result<String, Value> {
        self.key_value(self.line, model, None, sn, Some(ext))
    }

    fn key_sub_sn(&self, model: &str, sub: &str, sn: &str) -> Result<String, Value> {
        self.key_value(self.line, model, Some(sub), sn, None)
    }

    fn key_sub_sn_ext(&self, model: &str, sub: &str, sn: &str, ext: &str) -> Result<String, Value> {
        self.key_value(self.line, model, Some(sub), sn, Some(ext))
    }
}
